package com.small.nacos.config.server.service;

import com.small.nacos.common.notify.Event;
import com.small.nacos.common.notify.NotifyCenter;

import com.small.nacos.api.config.common.LongPullingConstants;
import com.small.nacos.common.notify.listener.Subscriber;
import com.small.nacos.config.server.model.event.LocalDataChangeEvent;
import com.small.nacos.config.server.utils.ConfigExecutor;
import com.small.nacos.config.server.utils.MD5Util;
import com.small.nacos.config.server.utils.RequestUtil;
import com.small.nacos.config.server.utils.ResponseUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.servlet.AsyncContext;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

@Service
public class LongPollingService {

    protected static final Logger LOGGER = LoggerFactory.getLogger(LongPollingService.class);

    private final int FIXED_DELAY_TIME = 500;
    //长轮询任务队列
    final Queue<ClientLongPolling> allSubs;

    public LongPollingService() {
        allSubs = new ConcurrentLinkedQueue<ClientLongPolling>();

        //注册配置文件改变订阅者
        NotifyCenter.registerSubscriber(new Subscriber() {
            @Override
            public void onEvent(Event event) {
                if (event instanceof LocalDataChangeEvent){
                    String groupKey = ((LocalDataChangeEvent) event).getGroupKey();
                    ConfigExecutor.executeLongPolling(new DataChangeTask(groupKey));
                }
            }
            @Override
            public Class<? extends Event> subscribeType() {
                return LocalDataChangeEvent.class;
            }
        });
    }

    public static boolean isSupportLongPolling(HttpServletRequest req) {
        return null != req.getHeader(LongPullingConstants.LONG_PULLING_TIMEOUT);
    }

    public void addLongPollingClient(HttpServletRequest req, HttpServletResponse rsp, Map<String, String> clientMd5Map,
                                     int probeRequestSize) {

        String str = req.getHeader(LongPullingConstants.LONG_PULLING_TIMEOUT);
        String noHangUpFlag = req.getHeader(LongPullingConstants.LONG_PULLING_TIMEOUT_NO_HANGUP);
        int delayTime = FIXED_DELAY_TIME;

        //为了防止超时 提前时间
        long timeout = Long.parseLong(str) - delayTime;

        //校验md5
        List<String> changedGroups = MD5Util.compareMd5(clientMd5Map);
        //说明有更新的配置
        if (changedGroups.size() > 0) {
            generateResponse(rsp, changedGroups);
            return;
            //说明最近加入的配置 需要立即返回
        } else if (noHangUpFlag != null && noHangUpFlag.equalsIgnoreCase("true")) {
            return;
        }

        String ip = RequestUtil.getRemoteIp(req);

        //必须由http线程调用，或发送响应
        final AsyncContext asyncContext = req.startAsync();
        //由自己控制超时时间
        asyncContext.setTimeout(0L);

        //执行长轮询任务
        ConfigExecutor.executeLongPolling(
                new ClientLongPolling(asyncContext, clientMd5Map, ip, probeRequestSize, timeout));
    }

    /**
     * 返回缓存
     *
     * @param response
     * @param changedGroups
     */
    void generateResponse(HttpServletResponse response, List<String> changedGroups) {
        if (null == changedGroups) {
            return;
        }
        try {
            final String respString = MD5Util.compareMd5ResultString(changedGroups);
            // 设置不允许缓存
            ResponseUtil.disableCache(response);
            response.setStatus(HttpServletResponse.SC_OK);
            response.getWriter().println(respString);
        } catch (Exception ex) {
            LOGGER.error(ex.toString(), ex);
        }
    }

    class ClientLongPolling implements Runnable {

        final AsyncContext asyncContext;

        final Map<String, String> clientMd5Map;

        final long createTime;

        final String ip;

        final int probeRequestSize;

        final long timeoutTime;

        Future<?> asyncTimeoutFuture;

        ClientLongPolling(AsyncContext ac, Map<String, String> clientMd5Map, String ip, int probeRequestSize,
                          long timeoutTime) {
            this.asyncContext = ac;
            this.clientMd5Map = clientMd5Map;
            this.probeRequestSize = probeRequestSize;
            this.createTime = System.currentTimeMillis();
            this.ip = ip;
            this.timeoutTime = timeoutTime;
        }

        @Override
        public void run() {
            asyncTimeoutFuture = ConfigExecutor.scheduleLongPolling(new Runnable() {
                @Override
                public void run() {
                    try {
                        boolean removeFlag = allSubs.remove(ClientLongPolling.this);
                        List<String> changedGroups = MD5Util.compareMd5(clientMd5Map);
                        if (removeFlag) {
                            //延时任务执行 说明配置没有改变
                            sendResponse(null);
                        } else {
                            LOGGER.warn("client subsciber's relations delete fail.");
                        }
                    } catch (Throwable t) {
                        LOGGER.error("long polling error:" + t.getMessage(), t.getCause());
                    }
                }
            }, timeoutTime, TimeUnit.MILLISECONDS);
            //添加进长轮询任务
            allSubs.add(this);
        }

        void sendResponse(List<String> changedGroups) {

            // 取消延时任务 直接完成请求
            if (null != asyncTimeoutFuture) {
                //传入false参数只能取消还没有开始的任务，若任务已经开始了，就任由其运行下去。
                asyncTimeoutFuture.cancel(false);
            }
            generateResponse(changedGroups);
        }

        void generateResponse(List<String> changedGroups) {
            if (changedGroups == null) {
                //让web容器完成刷新
                asyncContext.complete();
                return;
            }
            HttpServletResponse response = (HttpServletResponse) asyncContext.getResponse();
            try {
                final String respString = MD5Util.compareMd5ResultString(changedGroups);
                ResponseUtil.disableCache(response);
                response.setStatus(HttpServletResponse.SC_OK);
                response.getWriter().println(respString);
                asyncContext.complete();
            } catch (IOException ex) {
                LOGGER.error(ex.toString(), ex);
                asyncContext.complete();
            }
        }

    }

    class DataChangeTask implements Runnable {

        private String groupKey;

        public DataChangeTask(String groupKey) {
            this.groupKey = groupKey;
        }

        @Override
        public void run() {
            Iterator<ClientLongPolling> iterator = allSubs.iterator();
            while (iterator.hasNext()){
                ClientLongPolling longPolling = iterator.next();
                if (longPolling.clientMd5Map.containsKey(groupKey)) {
                    longPolling.sendResponse(Arrays.asList(groupKey));
                }
            }
        }
    }
}
